package com.flinksql.test;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;


/**
 * @author: Lin
 * @create: 2021-06-16 10:21
 * @description: 声明时间语义，基于事件时间，DDL写法
 **/
public class FlinkTableAPI_Test12_EvenTime {
    public static void main(String[] args) throws Exception {
        //1.建立环境，测试不设置CK
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //2.设置处理时间字段
        tableEnv.executeSql("create table sensor (id string" +
                ",ts bigint" +
                ",vc int" +
                ",t as to_timestamp(from_unixtime(ts/1000,'yyyy-MM-dd HH:mm:ss'))" +
                ",watermark for t as t - interval '5' second)" +
                "with("
                + "'connector' = 'filesystem',"
                + "'path' = 'input/sensor.txt',"
                + "'format' = 'csv')");
        Table table = tableEnv.from("sensor");
        table.printSchema();

    }
}
